

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using Confluent.Kafka;
    using NLog;
    using Stee.CAP8.MessageBus.Interfaces.Delegates;
    using Stee.CAP8.MessageBus.Interfaces.Models;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Factories;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Poll consumption implementation
    /// </summary>
    internal abstract class PollingBase : IDisposable
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        /// <summary>
        /// Timer interval for re-subscription
        /// </summary>
        protected const int TimerIntervalSubscribe = 5000;

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// JSON configuration file path for broker
        /// </summary>
        protected readonly string BrokerConfigurationJSON;

        /// <summary>
        /// Consumer configuration
        /// </summary>
        protected readonly ConsumerConfig ConsumerConfig;

        /// <summary>
        /// Kafka consumer builder
        /// </summary>
        protected ConsumerBuilder<string, byte[]> ConsumerBuilder;

        /// <summary>
        /// Kafka consumer
        /// </summary>
        protected IConsumer<string, byte[]> Consumer;

        /// <summary>
        /// Subscriber error handler
        /// </summary>
        protected readonly HandleBrokerError BrokerErrorHandler;

        /// <summary>
        /// Cancel token for Kafka consumer
        /// </summary>
        protected CancellationTokenSource CancellationToken;

        /// <summary>
        /// Mutex lock
        /// </summary>
        protected readonly object Mutex = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the message topic
        /// Mandatory configuration.
        /// </summary>
        internal TopicName TopicName { get; private set; }

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// default: '' importance: high
        /// Mandatory configuration.
        /// </summary>
        internal string GroupID
        {
            get { return (this.ConsumerConfig == null) ? string.Empty : this.ConsumerConfig.GroupId; }
        }

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID { get; protected set; } = string.Empty;

        /// <summary>
        /// Gets the flag indicating that subscriber is initialized and ready for reading messages
        /// </summary>
        public bool IsInitialized { get; protected set; } = false;

        /// <summary>
        /// Gets the flag indicating if consumer is currently polling for messages.
        /// This flag will set to FALSE if Consume method throws or the consumer library returns an error.
        /// </summary>
        public bool IsPolling { get; protected set; }

        /// <summary>
        /// Flag indicating polling task is cancelled
        /// </summary>
        public bool IsCancelled = false;

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed { get; protected set; } = false;

        /// <summary>
        /// Gets or sets the action handler for consumer error
        /// </summary>
        public Action<Error> HandleError { get; internal set; }

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the PollingImpl class
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="groupID">Subscriber group ID</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        protected PollingBase(string brokerConfigurationJSON, string groupID, HandleBrokerError brokerErrorHandler)
        {
            var configs = ConfigurationHelper.GetBrokerConfiguration(brokerConfigurationJSON, ConfigurationConstants.BrokerConfigSubscribeSection);
            this.BrokerErrorHandler = brokerErrorHandler ?? throw new ArgumentNullException($"Error handler for broker is null: {nameof(brokerErrorHandler)}");
            var group = ValidateHelper.CheckString(groupID, excpMsg: "Subscriber group ID cannot be null or empty.");

            if (configs?.Any() == true)
            {
                this.BrokerConfigurationJSON = brokerConfigurationJSON;
                var configuration = new SubscriberConfiguration(configs);

                this.ConsumerConfig = new ConsumerConfig
                {
                    BootstrapServers = configuration.BootstrapServers,
                    GroupId = group,
                    ClientId = configuration.ClientID,
                    AutoOffsetReset = configuration.AutoOffsetReset,
                    EnableAutoCommit = configuration.EnableAutoCommit,
                    EnableAutoOffsetStore = configuration.EnableAutoOffsetStore,
                    MaxPollIntervalMs = configuration.MaxPollIntervalMs,
                    SessionTimeoutMs = configuration.SessionTimeoutMs,
                    AutoCommitIntervalMs = configuration.AutoCommitIntervalMs,
                    CheckCrcs = configuration.CheckCrcs,
                    PartitionAssignmentStrategy = configuration.PartitionAssignmentStrategy,
                };

                //// Initialize consumer builder
                this.InitializeConsumer();
            }
            else
            {
                throw new ConfigurationErrorsException($"No configuration found for consumer in section: {ConfigurationConstants.BrokerConfigAdminSection}");
            }
        }

        /// <summary>
        /// Destructor
        /// </summary>
        ~PollingBase()
        {
            this.Dispose(false);
        }

        #endregion Constructors & Initializers


        #region Public Methods

        /// <summary>
        /// Initializes the consumer with topic for subscription
        /// </summary>
        /// <param name="topicName">Topic name for subscribing</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        internal bool Initialize(TopicName topicName)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Polling task is already initialized. IsInit: {this.IsInitialized}");
                }

                ValidateHelper.ValidateTopicName(topicName);
                this.TopicName = topicName;

                if (this.Consumer != null)
                {
                    this.Consumer.Subscribe(this.TopicName.Name);
                }

                return this.IsInitialized = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Start poll loop to consume message
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        internal void StartPolling()
        {
            try
            {
                Logger.Info("Polling task started!");
                this.IsPolling = true;

                this.CheckConsumerState();
                this.CancellationToken = new CancellationTokenSource();
                this.IsCancelled = false;

                //// Consumer poll loop
                while (this.IsPolling)
                {
                    this.Consume();
                }
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, "Subscription has been cancelled.");
                this.IsCancelled = true;
                this.IsPolling = false;
            }
            catch (KafkaRetriableException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsPolling = false;
            }
            catch (ConsumeException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsPolling = false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                this.IsPolling = false;
            }
        }

        /// <summary>
        /// Stop polling message
        /// </summary>
        internal void StopPolling()
        {
            try
            {
                if (this.CancellationToken != null && !this.CancellationToken.IsCancellationRequested)
                {
                    this.CancellationToken.Cancel();
                    this.CancellationToken.Dispose();
                }

                this.CancellationToken = null;
                this.IsCancelled = true;
                this.IsPolling = false;

                Logger.Info("Polling task stopped!");
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Commit message offset
        /// </summary>
        /// <param name="brokerMessage">Kafka cluster message, contains partition and offset information for commit</param>
        internal bool Commit(object brokerMessage)
        {
            try
            {
                if (brokerMessage == null)
                {
                    throw new ArgumentNullException($"Null broker message encountered: {nameof(brokerMessage)}");
                }
                else if (!(brokerMessage is ConsumeResult<string, byte[]>))
                {
                    throw new ArgumentException($"Invalid broker message type encountered: {brokerMessage.GetType().FullName}");
                }

                this.CheckConsumerState();

                var consumeResult = brokerMessage as ConsumeResult<string, byte[]>;
                this.Consumer.Commit(consumeResult);

                return true;
            }
            catch (KafkaException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.SubscribeErrorHandler(this.Consumer, ex.Error);
                return false;
            }
        }

        /// <summary>
        /// Reset consumer group offset to latest
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <returns>Return TRUE if the reset is successful else FALSE</returns>
        internal bool ResetOffsetToLatest(TopicName topicName)
        {
            try
            {
                ValidateHelper.ValidateTopicName(topicName);

                var partition = Partition.Any;
                var topicPartition = new TopicPartition(topicName.Name, partition);
                var offset = Offset.End;
                var topicPartitionOffset = new TopicPartitionOffset(topicPartition, offset);
                this.Consumer.Seek(topicPartitionOffset);

                return true;
            }
            catch (KafkaException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Dispose polling task
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Builds the consumer
        /// </summary>
        private void InitializeConsumer()
        {
            //// Note: If a key or value deserializer is not set, the 
            //// deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
            //// will be used automatically (where available). The default deserializer for string
            //// is UTF8. The default deserializer for Ignore returns null for all input data (including non-null data).
            this.ConsumerBuilder = new ConsumerBuilder<string, byte[]>(this.ConsumerConfig);
            this.ConsumerBuilder.SetKeyDeserializer(Deserializers.Utf8);
            this.ConsumerBuilder.SetValueDeserializer(Deserializers.ByteArray);
            this.ConsumerBuilder.SetErrorHandler(this.SubscribeErrorHandler);

            //// Build consumer
            lock (this.Mutex)
            {
                this.Consumer = this.ConsumerBuilder.Build();
            }
        }

        /// <summary>
        /// Check current polling state
        /// </summary>
        private void CheckConsumerState()
        {
            if (!this.IsInitialized)
            {
                throw new InvalidOperationException($"Polling task is not initialized. IsInit: {this.IsInitialized}");
            }
            else if (this.IsDisposed || !this.IsPolling || this.Consumer == null)
            {
                Logger.Error("Polling task have been disposed, create new instance required.");
                throw new TypeInitializationException(typeof(PollingBase).FullName, null);
            }
        }

        /// <summary>
        /// Consume/poll for messages
        /// </summary>
        private void Consume()
        {
            this.CancellationToken.Token.ThrowIfCancellationRequested();
            var consumeResult = this.Consumer.Consume(this.CancellationToken.Token);

            if (consumeResult == null)
            {
                Logger.Warn($"Kafka consumed result is null. Topic: {this.TopicName}, Group ID: {this.ConsumerConfig.GroupId}");
                return;
            }
            else if (consumeResult.IsPartitionEOF)
            {
                //// Reached end of topic X in partition Y at offset Z
                Logger.Info($"Reached end of topic: {consumeResult.Topic}, Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}.");
                return;
            }

            //// Check message expiry
            if (CheckMessageExpiry(consumeResult.Message.Timestamp.UtcDateTime, this.TopicName.ExpiryTime))
            {
                //// Publish message to DLQ
                var task = this.PublishDLQMessage(consumeResult);
                task.Wait(TimeSpan.FromSeconds(ConfigurationConstants.ProduceOperationTimeoutSeconds));

                if (task.Result && this.Commit(consumeResult))
                {
                    //// Commit message without handling/processing
                    Logger.Info("Expired message published to DLQ successfully.");
                }
                else
                {
                    //// Publish DLQ failed, did not commit
                    Logger.Warn($"Publish expired message failed. Topic: {this.TopicName}, Group ID: {this.ConsumerConfig.GroupId}");
                }
            }
            else
            {
                //// Extract message headers
                var properties = GetPropertiesFromHeader(consumeResult.Message.Headers);

                //// Handle message notification
                if (this.HandleMessage(consumeResult, properties) && this.Commit(consumeResult))
                {
                    Logger.Info("Message consumed successfully.");
                }
                else
                {
                    Logger.Warn($"Message consumption failed. Topic: {this.TopicName}, Group ID: {this.ConsumerConfig.GroupId}");
                }
            }
        }

        /// <summary>
        /// Publish consumed message to dead-letter topic
        /// </summary>
        /// <param name="consumeResult">Raw message from Kafka broker</param>
        private async Task<bool> PublishDLQMessage(ConsumeResult<string, byte[]> consumeResult)
        {
            var isPublished = false;
            var cancelToken = default(CancellationTokenSource);

            try
            {
                var topic = ConfigurationHelper.GenerateDLQTopicName(this.TopicName.Name);
                cancelToken = new CancellationTokenSource(TimeSpan.FromSeconds(ConfigurationConstants.ProduceOperationTimeoutSeconds));

                using (var producer = PublisherFactory.CreateNewPublisher(this.BrokerConfigurationJSON, this.PublishErrorHandler))
                {
                    var deliveryResult = await producer.ProduceAsync(topic, consumeResult.Message, cancelToken.Token).ConfigureAwait(false);

                    if (deliveryResult.Status == PersistenceStatus.NotPersisted)
                    {
                        //// Published message is not persisted/saved
                        Logger.Error($"Failed to persist DLQ message for topic \"{consumeResult.Topic}\". Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}");
                    }
                    else
                    {
                        isPublished = true;
                    }
                }
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"Failed to publish DLQ message for topic \"{consumeResult.Topic}\". Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}");
            }
            catch (TimeoutException ex)
            {
                Logger.Error(ex, $"Timeout on publishing DLQ message for topic \"{consumeResult.Topic}\". Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}");
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
            finally
            {
                cancelToken?.Dispose();
            }

            return isPublished;
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iconsumer">Consumer instance</param>
        /// <param name="error">Kafka error</param>
        private void SubscribeErrorHandler(IConsumer<string, byte[]> iconsumer, Error error)
        {
            try
            {
                this.BrokerErrorHandler?.Invoke(error.Code.ToString(), error.Reason, error.IsFatal);

                if (this.IsPolling && !this.IsCancelled && !this.IsDisposed)
                {
                    this.IsPolling = false;
                    this.HandleError(error);
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iproducer">Producer instance</param>
        /// <param name="error">Kafka error</param>
        private void PublishErrorHandler(IProducer<string, byte[]> iproducer, Error error)
        {
            if (error == null)
            {
                Logger.Error("No error returned for DLQ publisher error.");
            }
            else
            {
                Logger.Error($"Error encountered on publishing message: {error.Code}");
            }
        }

        /// <summary>
        /// Dispose consumer
        /// </summary>
        private void DisposeConsumer()
        {
            try
            {
                lock (this.Mutex)
                {
                    if (this.Consumer != null)
                    {
                        this.Consumer.Unsubscribe();
                        this.Consumer.Close();
                        this.Consumer.Dispose();

                        this.Consumer = null;
                        this.ConsumerBuilder = null;
                    }
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing">If disposing equals true, the method has been called directly
        /// or indirectly by a user's code. Managed and unmanaged resources can be disposed.
        /// If disposing equals false, the method has been called by the runtime from inside the
        /// finalizer and you should not reference other objects. Only unmanaged resources can be disposed.</param>
        protected virtual void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed && disposing)
                {
                    this.StopPolling();
                    this.DisposeConsumer();
                }

                this.IsDisposed = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Handle consumed messsage
        /// Related to US 238, priority
        /// https://docs.microsoft.com/en-us/dotnet/api/system.multicastdelegate?view=netcore-3.1
        /// </summary>
        /// <param name="consumeResult">Raw message from Kafka broker</param>
        /// <param name="properties">Message meta-data</param>
        /// <returns>Return TRUE if handled succesfully else FALSE</returns>
        protected abstract bool HandleMessage(ConsumeResult<string, byte[]> consumeResult, IDictionary<string, byte[]> properties);

        #endregion Private Methods


        #region Static Methods

        /// <summary>
        /// Check if message publish time has exceeded retention/expiry time
        /// </summary>
        /// <param name="utcDt">Message publish UTC date time</param>
        /// <param name="retentionTime"></param>
        /// <returns>Return TRUE if expired else FALSE</returns>
        private static bool CheckMessageExpiry(DateTime utcDt, TimeSpan retentionTime)
        {
            try
            {
                if (utcDt.Kind != DateTimeKind.Utc)
                {
                    Logger.Warn($"Failed to check message expiry, date time kind is not UTC. Kind: {utcDt.Kind}");
                }

                //// Note: DateTime does not convert UTC to local on subtraction
                var messageTimespan = DateTime.UtcNow.Subtract(utcDt);
                return messageTimespan.CompareTo(retentionTime) >= 0;
            }
            catch (ArgumentOutOfRangeException ex)
            {
                Logger.Error(ex, ex.Message);
                return true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Get properties from Kafka message headers
        /// </summary>
        /// <param name="headers">Kafka message headers</param>
        /// <returns>Return property collection mapping if available else NULL</returns>
        private static Dictionary<string, byte[]> GetPropertiesFromHeader(Headers headers)
        {
            Dictionary<string, byte[]> properties = null;

            if (headers != null)
            {
                properties = new Dictionary<string, byte[]>();
                var headerEnumtr = headers.GetEnumerator();

                while (headerEnumtr.MoveNext())
                {
                    var iHeader = headerEnumtr.Current;

                    //// Get header key
                    var key = iHeader.Key;

                    //// Get properties where key is NOT message type
                    if (!key.Equals(HeaderKeys.HeaderKeyMessageType, StringComparison.InvariantCultureIgnoreCase))
                    {
                        //// Get header value
                        properties[key] = iHeader.GetValueBytes();
                    }
                }
            }

            return properties;
        }

        /// <summary>
        /// De-serialize byte array in to string
        /// </summary>
        /// <param name="bytes">Byte array to de-serialize</param>
        /// <param name="encoding">Encoding type</param>
        /// <returns>Return deserialized byte arrary as string</returns>
        protected static string DeserializeString(byte[] bytes, Encoding encoding) =>
            (bytes == null || encoding == null) ? null : encoding.GetString(bytes);

        #endregion Static Methods
    }
}
